package com.atguigu.flink.sql.query;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by 黄凯 on 2023/6/27 0027 20:04
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 *
 * Deduplication语法:
 *  *    特殊的TopN语法. 特殊在 rk = 1
 */
public class Flink08_Deduplication {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Schema schema =
                Schema.newBuilder()
                        .column("user" , "string")
                        .column("url" , "string")
                        .column("ts", "bigint")
                        .columnByExpression("pt" ,"proctime()")
                        .columnByExpression("et" , "to_timestamp_ltz(ts, 3)")
                        .watermark("et" , "et - interval '0' second")
                        .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        table.printSchema();
        tableEnv.createTemporaryView("t1" , table);

        //求每个窗口中每个url最后到达的数据
        //1. 使用窗口补充窗口信息(window_start ,window_end)
        String windowSql =
                "select url,\n" +
                        "       et,\n" +
                        "       window_start,\n" +
                        "       window_end\n" +
                        "from table(\n" +
                        "        tumble(table t1 , descriptor(et), interval '10' second)\n" +
                        "    )";

        Table t2 = tableEnv.sqlQuery(windowSql);
//        t2.execute().print();
        tableEnv.createTemporaryView("t2",t2);

        //2. 基于每条数据的时间求排名
        String rankSql =
                "select url,\n" +
                        "       et,\n" +
                        "       window_start,\n" +
                        "       window_end,\n" +
                        "       row_number() over (partition by url ,window_start,window_end order by et desc ) rk\n" +
                        "from t2";

        Table t3 = tableEnv.sqlQuery(rankSql);
        tableEnv.createTemporaryView("t3",t3);

        //3. 取topn
        String topnSql =
                "select url,\n" +
                        "       et,\n" +
                        "       window_start,\n" +
                        "       window_end,\n" +
                        "       rk\n" +
                        "from t3\n" +
                        "where rk = 1";

        tableEnv.sqlQuery(topnSql).execute().print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
